#!/usr/bin/python

import sys,os,socket,time
from threading import Thread

def deleteTables(machineName, psqldir, port, dbname, tables):
  tableList = tables.split()
  dropTables = ""
  for table in tableList:
    dropTables += "drop table " + table + ";"
  os.system("ssh %s '%spsql -p %s -d %s -c \"%s\"'" % (machineName, psqldir, port, dbname, dropTables))

def execRemoteQuery(machineName, psqldir, port, dbname, outputdir, query):
  os.system("ssh %s '%spsql -p %s -d %s -c \"%s\" -t -o %s-%s -A -F \"\t\" ' " % (machineName, psqldir, port, dbname, query, outputdir, machineName))

def execRemoteQueryWithAFile(machineName, psqldir, port, dbname, outputdir, queryfile):
  os.system("ssh %s '%spsql -p %s -d %s -f %s -t -o %s-%s -A -F \"\t\" ' " % (machineName, psqldir, port, dbname, queryfile, outputdir, machineName))

def broadcastOutput(machineName, allnodes, fname, TMP_DIR):
  # Rename the files to include 'intermediate' so that they can be easily concatenated
  os.system("ssh %s ' mv %s-%s %s-intermediate-%s;'" % (machineName, fname, machineName, fname, machineName))    
  fname = fname + "-intermediate-" + machineName
  for anode in allnodes:
    anode = anode.strip()
    if anode == machineName:
      continue
    try:
      os.system("ssh %s 'scp %s %s:%s;'" % (machineName, fname, anode, TMP_DIR))
    except IOError:
      print "No such file?"

def importData(machineName, CODE_DIR, TMP_DIR, psqldir, port, dbname, table, schema):
  #print ("ssh %s '%simport.py \"%s\" \"%s\" \"%s\" \"%s\" %s %s %st1-* %st2-*'" % (machineName, CODE_DIR, CODE_DIR, psqldir, port, dbname, createT1, createT2, TMP_DIR, TMP_DIR))
  #os.system ("ssh %s '%simport.py \"%s\" \"%s\" \"%s\" \"%s\" %s %s %st1-* %st2-*; rm -f %s/t*-*'" % (machineName, CODE_DIR, CODE_DIR, psqldir, port, dbname, createT1, createT2, TMP_DIR, TMP_DIR, TMP_DIR))
  files = TMP_DIR + table + "-*"
  os.system ("ssh %s '%simport.py \"%s\" \"%s\" \"%s\" \"%s\" \"%s\" %s %s; rm -f %s'" % (machineName, CODE_DIR, CODE_DIR, psqldir, port, dbname, table, schema, files, files))

def hash(machineName, CODE_DIR, fname, numMachines, hashColumn):
  fname = fname + "-" + machineName
  os.system("ssh %s 'cd %s; java Hasher \"%s\" %s %s; rm -f %s'" % (machineName, CODE_DIR, fname, numMachines, hashColumn, fname))
  #os.system("ssh %s 'cd %s; java Hasher \"%s\" %s %s;'" % (machineName, CODE_DIR, fname, numMachines, hashColumn))

def moveInOneMachine(hostfile, prefix, TMP_DIR):
  #print("moveInOneMachine")
  nodes = open(hostfile, 'r').readlines()
  threads = []
  for i in range(1, len(nodes)+1):
    node = nodes[int(i)-1].strip()
    if node == socket.gethostname():
      continue
    file = prefix + str(i)
    t = Thread(target=moveAndDelete, args=(file, node, TMP_DIR))
    #print("%s   %s" % (file, node))
    threads.append(t)
    t.start()
  for thread in threads:
    thread.join  

def moveAndDelete(file, dest, TMP_DIR):
   #print("scp %s %s:%s; rm -f %s" % (file, dest, TMP_DIR, file))
   #start_time = time.time()
   #os.system("ssh %s 'sleep 10'" % (src))
   #total = time.time() - start_time
   #print ("%s %s" % (file, total))
   os.system("scp %s %s:%s; rm -f %s" % (file, dest, TMP_DIR, file))
   #total = time.time() - start_time
   #print ("%s %s" % (file, total))

def move(node, hostfile, CODE_DIR, prefix, TMP_DIR):
  prefix = prefix + "-" + node + "-intermediate-"
  #print("ssh %s 'cd %s; python join.py moveInOneMachine %s %s %s'" % (node, CODE_DIR, hostfile, prefix, TMP_DIR))
  os.system("ssh %s 'cd %s; python join.py moveInOneMachine %s %s %s'" % (node, CODE_DIR, hostfile, prefix, TMP_DIR))

if __name__ == '__main__':
  hostfile = sys.argv[2]
  nodes = open(hostfile, 'r').readlines()

  if sys.argv[1] == "moveInOneMachine":
    #start_time = time.time()
    #print("moveInOneMachine haha ")
    moveInOneMachine(hostfile, sys.argv[3], sys.argv[4])
    #print ("Total execution: %s" % (time.time() - start_time))
    sys.exit()

  threads = []
  for node in nodes:
    node = node.strip()
    if sys.argv[1] == "LOCAL_JOIN":
      t = Thread(target=execRemoteQuery, args=(node, sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6], sys.argv[7]))
      threads.append(t)
      t.start()
    if sys.argv[1] == "LOCAL_JOIN_WITH_A_FILE":
      t = Thread(target=execRemoteQueryWithAFile, args=(node, sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6], sys.argv[7]))
      threads.append(t)
      t.start()
    if sys.argv[1] == "HASH":
      t = Thread(target=hash, args=(node, sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6]))
      threads.append(t)
      t.start()
    if sys.argv[1] == "MOVE":
      t = Thread(target=move, args=(node, hostfile, sys.argv[3], sys.argv[4], sys.argv[5]))
      threads.append(t)
      t.start()
    if sys.argv[1] == "BROADCAST":
      t = Thread(target=broadcastOutput, args=(node, nodes, sys.argv[3], sys.argv[4]))
      threads.append(t)
      t.start()
    if sys.argv[1] == "IMPORT":
      t = Thread(target=importData, args=(node, sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6], sys.argv[7], sys.argv[8], sys.argv[9]))
      threads.append(t)
      t.start()
    if sys.argv[1] == "DELETE_TABLES":
      t = Thread(target=deleteTables, args=(node, sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6]))
      threads.append(t)
      t.start()
  for thread in threads:
    thread.join
